package com.pc.cloud.message;

import com.google.code.yanf4j.core.impl.StandardSocketOption;
import com.pc.fqueue.memcached.FqueueTranscoder;
import com.pc.fqueue.memcached.StartNewQueue;
import com.pc.fqueue.util.Config;
import com.pc.fqueue.util.QueueUtils;
import net.rubyeye.xmemcached.MemcachedClient;
import net.rubyeye.xmemcached.MemcachedClientBuilder;
import net.rubyeye.xmemcached.XMemcachedClientBuilder;
import net.rubyeye.xmemcached.exception.MemcachedException;
import net.rubyeye.xmemcached.transcoders.Transcoder;
import net.rubyeye.xmemcached.utils.AddrUtil;
import org.json.JSONException;

import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;

public class MessageManager {
	// private Map<String, MemcachedClient> sendClients;

	private MemcachedClient sendClent;

	// private String consumeKey;

	//private Map<String, MemcachedClient> consumeClients;
	
	private Map<String, Listener> listeners;

	public void setListeners(Map<String, Listener> listeners) {
		this.listeners = listeners;
	}

	private static final Transcoder transcoder = new FqueueTranscoder();

	private void init() {

		StartNewQueue.newQueueInstance(Integer.parseInt(Config.getSetting("port")));

		// send
		this.sendClent = this.getClient("127.0.0.1:" + Config.getSetting("port"));

		// consume
		String ipsline = Config.getSetting("clusterIPs");
		String[] ips = QueueUtils.parse(ipsline, '|');
		ExecutorService pool = Executors.newFixedThreadPool(this.listeners.size() * ips.length);
		for(Entry<String,Listener> entry : this.listeners.entrySet()) {
			String queue = entry.getKey();
			Listener listener = entry.getValue();
			int no = 1;
			for(String ip : ips) {
				StringBuilder key = new StringBuilder();
				key.append(queue);
				key.append(no);
				key.append("_");
				key.append(Config.getSetting("password"));
				pool.execute(new ReceiveMessageWorker(key.toString(),this.getClient(ip),listener));
				no++;
			}
		}

	}

	public boolean sendMessage(String queue, String message)
			throws TimeoutException, InterruptedException, MemcachedException,
			JSONException {
		int totalQueues = Config.getInt("totalInstances");
		for (int i = 1; i <= totalQueues; i++) {
			StringBuilder key = new StringBuilder();
			key.append(queue);
			key.append(i);
			key.append("_");
			key.append(Config.getSetting("password"));
			this.sendClent.set(key.toString(), 0, message,transcoder);
		}
		return true;
	}

	private MemcachedClient getClient(String address) {

		// "127.0.0.1:12001"
		MemcachedClientBuilder builder = new XMemcachedClientBuilder(
				AddrUtil.getAddresses(address));
		builder.setConnectionPoolSize(50); // set connection pool size to five

		try {
			MemcachedClient client = builder.build();
			client.setOptimizeGet(false);
			builder.setSocketOption(StandardSocketOption.SO_KEEPALIVE, true);
			builder.setSocketOption(StandardSocketOption.SO_RCVBUF, 64 * 1024);
			builder.setSocketOption(StandardSocketOption.SO_SNDBUF, 64 * 1024);
			builder.setSocketOption(StandardSocketOption.SO_REUSEADDR, true);
			builder.setSocketOption(StandardSocketOption.TCP_NODELAY, false);

			client.setOpTimeout(5000);

			return client;
		} catch (Exception e) {
			// e.printStackTrace();
			System.out.println(e.getMessage());
		}
		return null;
	}

	public class ReceiveMessageWorker implements Runnable {
		private String key;
		private MemcachedClient client;
		private Listener listener;

		public ReceiveMessageWorker(String key, MemcachedClient value,
				Listener listener) {
			this.key = key;
			this.client = value;
			this.listener = listener;
		}

		@Override
		public void run() {
			while (true) {
				try {
					try {
						String message = (String)client.get(key, transcoder);
						while (message != null) {
							try {
								this.listener.onMessage(message);
							} catch (Exception e) {
								e.printStackTrace();
							} finally {
								client.delete(key);
							}

							message = client.get(key);
						}
					} catch (TimeoutException e) {
						e.printStackTrace();
					} catch (MemcachedException e) {
						e.printStackTrace();
					}
					Thread.sleep(1000*5);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}

	}
}
